package kafkaserver

import (
	"context"
	"core/kafka"
	"core/server/pushserver"
	"encoding/json"
	"github.com/dubbogo/gost/log/logger"
	kafkago "github.com/segmentio/kafka-go"
)

type PushServer struct {
	Name  string
	Topic string
	pushserver.PushServer
	writer *kafkago.Writer
}

func (p *PushServer) Start() {

	p.writer = kafka.GetKafkaWriter(p.Topic)
	p.PushServer.Name = p.Name
	p.PushServer.Do = p.Do
	p.PushServer.Start()
}

func (p *PushServer) Do() int {
	var r int
	for {
		var l = p.PopAll()

		if len(l) == 0 {
			break
		}
		r += len(l)
		var lmsg []kafkago.Message
		for _, v := range l {
			lmsg = append(lmsg, v.(kafkago.Message))
		}
		var err = p.writer.WriteMessages(context.Background(),
			lmsg...,
		)
		if err != nil {
			logger.Errorf("发送kafka数据失败：%v", err)
		}

		//logger.Errorf("发送kafka数据数量：%v", len(l))
	}
	return r
}

func (p *PushServer) Push(key string, elem interface{}) error {
	var body, err = json.Marshal(elem)
	if err != nil {
		return err
	}
	var m = kafkago.Message{
		Key:   []byte(key),
		Value: body,
	}
	p.PushServer.Push(m)
	return nil
}
